package com.dabai.iot.core.service.impl;

import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.RandomUtil;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.dabai.common.core.exception.ServiceException;
import com.dabai.common.core.utils.Func;
import com.dabai.common.core.utils.ObjectUtils;
import com.dabai.common.redis.utils.RedisUtils;
import com.dabai.common.tenant.helper.TenantHelper;
import com.dabai.iot.core.cache.CacheKey;
import com.dabai.iot.core.dto.ConsumerGroupInput;
import com.dabai.iot.core.dto.ConsumerGroupUpdate;
import com.dabai.iot.core.entity.ConsumerGroupEntity;
import com.dabai.iot.core.entity.ServerSubscribeConsumerEntity;
import com.dabai.iot.core.entity.ServerSubscribeEntity;
import com.dabai.iot.core.mapper.ConsumerGroupMapper;
import com.dabai.iot.core.mapper.ServerSubscribeMapper;
import com.dabai.iot.core.service.IConsumerGroupService;
import com.dabai.iot.core.service.IServerSubscribeConsumerService;
import com.github.yulichang.toolkit.JoinWrappers;
import com.github.yulichang.wrapper.MPJLambdaWrapper;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.List;

/**
 * 消费组 Service 实现类
 *
 * @author kev1n
 */
@Service
@RequiredArgsConstructor
public class ConsumerGroupServiceImpl extends ServiceImpl<ConsumerGroupMapper, ConsumerGroupEntity> implements IConsumerGroupService {

    private final IServerSubscribeConsumerService serverSubscribeConsumerService;
    private final ServerSubscribeMapper serverSubscribeMapper;

    @Override
    @Transactional(rollbackFor = Exception.class)
    public boolean addConsumer(ConsumerGroupInput input) {
        if (checkConsumerNameDuplicate(null, input.getGroupName())) {
            throw new ServiceException("已经存在相同的消费组名称");
        }

        ConsumerGroupEntity consumerGroup = new ConsumerGroupEntity();
        consumerGroup.setGroupName(input.getGroupName());

        // groupID 全系统唯一
        List<ConsumerGroupEntity> groupEntities = TenantHelper.ignore(() -> this.list());
        List<String> groupIDs = groupEntities.stream().map(ConsumerGroupEntity::getGroupIdentifier).toList();
        String groupID = generateGroupID(groupIDs);
        consumerGroup.setGroupIdentifier(groupID);

        boolean result = save(consumerGroup);
        if (result) {
            result = submitSubscribeConsumer(consumerGroup.getId(), input.getSubscribeIds());
            if (!result) {
                // 创建消费组失败
                throw new ServiceException("创建消费组失败");
            }
        }
        return result;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public boolean updateConsumer(ConsumerGroupUpdate update) {
        if (checkConsumerNameDuplicate(update.getId(), update.getGroupName())) {
            throw new ServiceException("已经存在相同的消费组名称");
        }

        ConsumerGroupEntity consumerGroup = new ConsumerGroupEntity();
        consumerGroup.setId(update.getId());
        consumerGroup.setGroupName(update.getGroupName());
        consumerGroup.setGroupIdentifier(null); // 标识符不允许修改

        boolean result = saveOrUpdate(consumerGroup);
        if (result) {
            result = submitSubscribeConsumer(consumerGroup.getId(), update.getSubscribeIds());
            if (!result) {
                // 创建消费组失败
                throw new ServiceException("创建消费组失败");
            }
        }
        return result;
    }

    /**
     * 校验消费组名称是否重复(企业内)
     */
    boolean checkConsumerNameDuplicate(Long id, String groupName) {
        return this.lambdaQuery()
            .eq(ConsumerGroupEntity::getGroupName, groupName)
            .ne(ObjectUtil.isNotNull(id), ConsumerGroupEntity::getId, id)
            .exists();
    }

    /**
     * 建立消费组与订阅的关系
     *
     * @param consumerId   消费组Id
     * @param subscribeIds 选择的订阅列表Ids
     */
    private boolean submitSubscribeConsumer(Long consumerId, List<Long> subscribeIds) {
        // 先删除与消费组关联的所有订阅关系
        removeSubscribeConsumerRelation(consumerId);

        // 再建立新关系
        List<ServerSubscribeConsumerEntity> newRelationList = subscribeIds.stream().map(subscribeId -> {
            ServerSubscribeEntity serverSubscribeEntity = serverSubscribeMapper.selectById(subscribeId);
            if (serverSubscribeEntity != null) {
                ServerSubscribeConsumerEntity sc = new ServerSubscribeConsumerEntity();
                sc.setProductId(serverSubscribeEntity.getProductId());
                sc.setSubscribeId(subscribeId);
                sc.setConsumerId(consumerId);
                return sc;
            }
            return null;
        }).toList();
        boolean result = serverSubscribeConsumerService.saveBatch(newRelationList);

        if (result) {
            for (ServerSubscribeConsumerEntity sc : newRelationList) {
                Long productId = sc.getProductId();
                deleteCacheByProductId(productId);
            }
        } else {
            throw new ServiceException("操作失败");
        }

        return true;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public boolean deleteConsumer(@NonNull Long id) {
        ConsumerGroupEntity consumerGroup = getById(id);
        if (consumerGroup == null) {
            throw new ServiceException("消费组不存在");
        }

        boolean result = removeById(id);
        if (result) {
            removeSubscribeConsumerRelation(id);
            deleteCache(consumerGroup.getGroupIdentifier());
        }
        return result;
    }

    /**
     * 删除缓存
     */
    private void deleteCache(@NonNull String groupID) {
        TenantHelper.ignore(() -> RedisUtils.deleteObject(CacheKey.getConsumerKey(groupID)));
    }

    @Override
    public IPage<ConsumerGroupEntity> getConsumerGroupPage(IPage<ConsumerGroupEntity> page, ConsumerGroupEntity consumerGroup) {
        QueryWrapper<ConsumerGroupEntity> queryWrapper = new QueryWrapper<>();
        queryWrapper.lambda()
            .eq(Func.isNotEmpty(consumerGroup.getId()), ConsumerGroupEntity::getId, consumerGroup.getId())
            .eq(Func.isNotEmpty(consumerGroup.getTenantId()), ConsumerGroupEntity::getTenantId, consumerGroup.getTenantId())
            .like(Func.isNotEmpty(consumerGroup.getGroupName()), ConsumerGroupEntity::getGroupName, consumerGroup.getGroupName())
            .eq(Func.isNotEmpty(consumerGroup.getGroupIdentifier()), ConsumerGroupEntity::getGroupIdentifier, consumerGroup.getGroupIdentifier())
            .orderByDesc(ConsumerGroupEntity::getCreateTime);

        return page(page, queryWrapper);
    }

    @Override
    public ConsumerGroupEntity getByGroupID(@NonNull String groupID) {
        return this.lambdaQuery()
            .eq(ConsumerGroupEntity::getGroupIdentifier, groupID)
            .one();
    }

    @Override
    public ConsumerGroupEntity getCacheByGroupID(@NonNull String groupID) {
        String cacheKey = CacheKey.getConsumerKey(groupID);
        ConsumerGroupEntity consumerGroupCache = TenantHelper.ignore(() -> RedisUtils.getCacheObject(cacheKey));
        if (consumerGroupCache != null) {
            return consumerGroupCache;
        }

        final ConsumerGroupEntity consumerGroup = this.getByGroupID(groupID);
        if (consumerGroup != null) {
            consumerGroupCache = consumerGroup;
            TenantHelper.ignore(() -> RedisUtils.setCacheObject(cacheKey, consumerGroup));
        }

        return consumerGroupCache;
    }

    @Override
    public List<ConsumerGroupEntity> getBySubscribeId(@NonNull Long subscribeId) {
        MPJLambdaWrapper<ConsumerGroupEntity> wrapper = JoinWrappers.lambda(ConsumerGroupEntity.class)
            .selectAll(ConsumerGroupEntity.class)
            .innerJoin(ServerSubscribeConsumerEntity.class, ServerSubscribeConsumerEntity::getConsumerId, ConsumerGroupEntity::getId)
            .innerJoin(ServerSubscribeEntity.class, on -> on
                .eq(ServerSubscribeEntity::getId, ServerSubscribeConsumerEntity::getSubscribeId)
                .eq(ServerSubscribeEntity::getId, subscribeId)
            );

        return baseMapper.selectJoinList(ConsumerGroupEntity.class, wrapper);
    }

    @Override
    public List<ConsumerGroupEntity> getByProductId(@NonNull Long productId) {
        MPJLambdaWrapper<ConsumerGroupEntity> wrapper = JoinWrappers.lambda(ConsumerGroupEntity.class)
            .selectAll(ConsumerGroupEntity.class)
            .innerJoin(ServerSubscribeConsumerEntity.class, ServerSubscribeConsumerEntity::getConsumerId, ConsumerGroupEntity::getId)
            .innerJoin(ServerSubscribeEntity.class, on -> on
                .eq(ServerSubscribeEntity::getId, ServerSubscribeConsumerEntity::getSubscribeId)
                .eq(ServerSubscribeEntity::getProductId, productId)
            );

        return baseMapper.selectJoinList(ConsumerGroupEntity.class, wrapper);
    }

    @Override
    public List<ConsumerGroupEntity> getCacheByProductId(Long productId) {
        String cacheKey = CacheKey.getConsumerListKey(productId);
        List<ConsumerGroupEntity> groupListCache = TenantHelper.ignore(() -> RedisUtils.getCacheList(cacheKey));
        if (ObjectUtils.isNotEmpty(groupListCache)) {
            return groupListCache;
        }

        List<ConsumerGroupEntity> groupList = this.getByProductId(productId);
        if (ObjectUtils.isNotEmpty(groupList)) {
            groupListCache = groupList;
            TenantHelper.ignore(() -> RedisUtils.setCacheList(cacheKey, groupList));
        }
        return groupListCache;
    }

    @Override
    public boolean unsubscription(@NonNull Long consumerId, @NonNull Long serverSubscribeId) {
        Long count = serverSubscribeConsumerService.lambdaQuery()
            .eq(ServerSubscribeConsumerEntity::getConsumerId, consumerId)
            .count();

        if (count < 2) {
            // 至少要保留一个订阅产品
            throw new ServiceException("至少要保留一个订阅产品");
        }

        LambdaUpdateWrapper<ServerSubscribeConsumerEntity> wrapper = Wrappers.<ServerSubscribeConsumerEntity>update().lambda()
            .eq(ServerSubscribeConsumerEntity::getConsumerId, consumerId)
            .eq(ServerSubscribeConsumerEntity::getSubscribeId, serverSubscribeId);
        boolean remove = serverSubscribeConsumerService.remove(wrapper);
        if (remove) {
            ServerSubscribeEntity serverSubscribeEntity = serverSubscribeMapper.selectById(serverSubscribeId);
            if (serverSubscribeEntity != null) {
                this.deleteCacheByProductId(serverSubscribeEntity.getProductId());
            }
        } else {
            throw new ServiceException("解除订阅失败");
        }
        return true;
    }

    /**
     * 删除与消费组关联的所有订阅关系
     *
     * @param consumerId 消费组Id
     */
    private void removeSubscribeConsumerRelation(Long consumerId) {
        // 记录下关联的产品Id
        List<Long> productIds = serverSubscribeConsumerService.lambdaQuery()
            .eq(ServerSubscribeConsumerEntity::getConsumerId, consumerId)
            .list()
            .stream()
            .map(ServerSubscribeConsumerEntity::getProductId)
            .toList();
        if (ObjectUtils.isEmpty(productIds)) {
            return;
        }

        LambdaUpdateWrapper<ServerSubscribeConsumerEntity> wrapper = Wrappers.<ServerSubscribeConsumerEntity>update().lambda()
            .eq(ServerSubscribeConsumerEntity::getConsumerId, consumerId);
        boolean remove = serverSubscribeConsumerService.remove(wrapper);
        if (remove) {
            for (Long productId : productIds) {
                this.deleteCacheByProductId(productId);
            }
        } else {
            throw new ServiceException("删除与消费组关联的所有订阅关系失败");
        }
    }

    /**
     * 删除产品Id关联的消费组缓存
     */
    private void deleteCacheByProductId(Long productId) {
        String cacheKey = CacheKey.getConsumerListKey(productId);
        TenantHelper.ignore(() -> RedisUtils.deleteObject(cacheKey));
    }

    /**
     * 生成 GroupID
     */
    private String generateGroupID(List<String> groupIDs) {
        String groupID = RandomUtil.randomString(12);
        if (groupIDs.contains(groupID)) {
            return generateGroupID(groupIDs);
        }
        return groupID;
    }

}
